package com.dbms.execution.execution;

import com.dbms.common.Schema;
import com.dbms.common.Tuple;
import com.dbms.execution.plan.PlanNode;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * HashJoinExecutor (哈希连接执行器) 实现了高效的哈希连接算法。
 * 这种算法分为两个阶段：
 * 1. 构建阶段 (Build Phase): 将一个较小的表（内表/inner table）完全读入内存，
 * 并根据连接键（join key）构建一个哈希表。
 * 2. 探测阶段 (Probe Phase): 逐行读取较大的表（外表/outer table），
 * 并使用其连接键在哈希表中快速查找匹配的行。
 * 它假设内表（build side）可以完全加载到内存中。
 */
public class HashJoinExecutor implements Executor {

    /**
     * 外表执行器 (Probe side)，通常是数据量较大的表。
     */
    private final Executor outerExecutor;
    /**
     * 内表执行器 (Build side)，通常是数据量较小的表，用于构建哈希表。
     */
    private final Executor innerExecutor;
    /**
     * 连接谓词，定义了连接条件（例如 `tableA.id = tableB.a_id`）。
     */
    private final Predicate joinPredicate;
    /**
     * 连接操作后输出的元组模式。
     */
    private final Schema outputSchema;

    /**
     * 内存中的哈希表，用于存储构建表（内表）的数据。
     * Key 是连接键的值，Value 是具有相同连接键的元组列表。
     */
    private Map<Object, List<Tuple>> hashTable;
    /**
     * 当前正在处理的外表元组。
     */
    private Tuple outerTuple;
    /**
     * 当前外表元组在哈希表中匹配到的内表元组列表。
     */
    private List<Tuple> innerMatches;
    /**
     * 指向 `innerMatches` 列表中下一个待处理元组的索引。
     */
    private int innerMatchIndex;

    public HashJoinExecutor(Executor outerExecutor, Executor innerExecutor, Predicate predicate, Schema outputSchema) {
        this.outerExecutor = outerExecutor;
        this.innerExecutor = innerExecutor;
        this.joinPredicate = predicate;
        this.outputSchema = outputSchema;
        this.hashTable = new HashMap<>();
        this.outerTuple = null;
    }

    @Override
    public void open() {
        // --- 构建阶段 (Build Phase) ---
        // 1. 打开内表执行器，准备读取数据
        innerExecutor.open();
        while (true) {
            Tuple tuple = innerExecutor.next();
            if (tuple == null) {
                break; // 内表已全部读取完毕
            }
            // 2. 从内表元组中获取连接键的值
            Object key = tuple.getValue(joinPredicate.getRightFieldIndex());
            // 3. 将元组存入哈希表。如果多个元组有相同的键，它们会被放入同一个列表中
            hashTable.computeIfAbsent(key, k -> new ArrayList<>()).add(tuple);
        }
        innerExecutor.close(); // 构建完成，关闭内表执行器

        // --- 初始化探测阶段 (Probe Phase) ---
        outerExecutor.open();
        this.outerTuple = null;
        this.innerMatches = null;
        this.innerMatchIndex = 0;
    }

    @Override
    public Tuple next() {
        // --- 探测阶段 (Probe Phase) ---
        while (true) {
            // 1. 检查是否需要获取新的外表元组。
            //    (条件：还没有外表元组，或者当前外表元组的所有匹配项都已处理完)
            if (outerTuple == null || innerMatches == null || innerMatchIndex >= innerMatches.size()) {
                // 1a. 从外表获取下一个元组
                outerTuple = outerExecutor.next();
                if (outerTuple == null) {
                    // 如果外表已经遍历完，则整个连接过程结束
                    return null;
                }
                // 1b. 使用外表元组的连接键去探测哈希表
                Object key = outerTuple.getValue(joinPredicate.getLeftFieldIndex());
                innerMatches = hashTable.get(key); // 获取匹配的内表元组列表
                innerMatchIndex = 0; // 重置匹配列表的索引
            }

            // 2. 检查是否有匹配项
            if (innerMatches != null) {
                // 2a. 如果有匹配，从匹配列表中取出一个内表元组
                Tuple innerTuple = innerMatches.get(innerMatchIndex);
                innerMatchIndex++; // 移动到下一个匹配项，为下次调用做准备

                // 2b. 将外表元组和内表元组的值组合成一个新的结果元组
                List<Object> combinedValues = new ArrayList<>();
                for (int i = 0; i < outerExecutor.getSchema().getColumnCount(); i++) {
                    combinedValues.add(outerTuple.getValue(i));
                }
                for (int i = 0; i < innerExecutor.getSchema().getColumnCount(); i++) {
                    combinedValues.add(innerTuple.getValue(i));
                }
                return new Tuple(combinedValues, outputSchema);
            }
            // 3. 如果没有匹配 (innerMatches == null)，则当前 outerTuple 没有匹配项。
            //    循环会继续，在下一次迭代的开始处获取下一个 outerTuple。
        }
    }

    @Override
    public void close() {
        outerExecutor.close();
        innerExecutor.close();
        hashTable.clear(); // 释放哈希表占用的内存
    }

    @Override
    public Schema getSchema() {
        return outputSchema;
    }

    @Override
    public PlanNode getPlanNode() {
        return null;
    }

    // 为了让Predicate能同时被NestedLoopJoin和HashJoin使用，我们需要添加getter
    // 请确保您的 Predicate.java 中有这两个方法
    // public int getLeftFieldIndex() { return leftFieldIndex; }
    // public int getRightFieldIndex() { return rightFieldIndex; }
}